package com.epfly.promo.activity.impl;


import com.epfly.base.ex.PromoException;
import com.epfly.base.ex.PromoMessageEnum;
import com.epfly.base.id.SnowflakeIdService;
import com.epfly.base.util.DateUtil;
import com.epfly.iface.promo.activity.ActivityTaskService;
import com.epfly.iface.promo.activity.enums.ActivityExecResultEnum;
import com.epfly.iface.promo.activity.enums.ActivityStatusEnum;
import com.epfly.iface.promo.activity.enums.ActivityTaskTypeEnum;
import com.epfly.iface.promo.activity.enums.IsActiveEnum;
import com.epfly.ifood.base.platform.model.dto.SysAlarmInfo;
import com.epfly.promo.activity.dao.ActivityActiveStoredMapper;
import com.epfly.promo.activity.dao.ActivityMapper;
import com.epfly.promo.activity.dao.ActivityStoreMapper;
import com.epfly.promo.activity.dao.ActivityTaskMapper;
import com.epfly.promo.activity.entity.ActivityActiveStoreEntity;
import com.epfly.promo.activity.entity.ActivityEntity;
import com.epfly.promo.activity.entity.ActivityStoreEntity;
import com.epfly.promo.activity.entity.ActivityTaskEntity;
import com.epfly.promo.activity.feign.SysAlarmServiceFeign;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.context.scope.refresh.RefreshScopeRefreshedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.stream.Collectors;

@Slf4j
@Service
public class ActivityTaskServiceImpl implements ActivityTaskService {


    /**
     * 构造方法注入，便于初始化线程池
     */
    private final ActivityPoolConfiguration activityPoolConfiguration;

    /**
     * 线程池
     */
    private ThreadPoolExecutor executor;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private SysAlarmServiceFeign sysAlarmService;

    @Resource
    private ActivityTaskMapper activityTaskMapper;

    @Resource
    private ActivityMapper activityMapper;

    @Resource
    private ActivityStoreMapper activityStoreMapper;


    @Resource
    private ActivityActiveStoredMapper activityActiveStoredMapper;

    @Resource
    private SnowflakeIdService snowflakeIdService;

    @Resource
    private ActivitySupport activitySupport;


    public ActivityTaskServiceImpl(@Autowired ActivityPoolConfiguration activityPoolConfiguration) {
        this.activityPoolConfiguration = activityPoolConfiguration;
        createPool(activityPoolConfiguration);
    }


    /**
     * 定时使活动生效
     * note：
     * 任务的执行状态被置为执行中，即将进入线程池的任务，在没有执行结束前，被意外终止。此时会提供超时机制；
     * 比如，当前时间大于等于提交到线程池的时间30分钟，根据执行次数重置任务的执行状态为待执行或执行失败；
     * 这样生产者就可以重新扫描到此任务，并加入线程池重新执行。
     * <p>
     * !!!
     * 但是超时机制没法去区分，是意外终止还是某个任务执行时间执行超过了半个小时。如果是后者，会导致任务被执行多次，而占用了线程池的线程资源。
     * 在任务开始前，记录当前任务ID，表示此任务正在执行。
     * 在finally代表块，清理此ID。
     * 线程池由于异常终止，重新生动时，清理正在执行的任务ID集合。
     * <p>
     * execResult = "RUNNING"
     * 只能表示任务的执行状态被置为执行中，将提交到线程池，至于一段时间后，此任务真实状态未知。
     * <p>
     * <p>
     * 相关方法：
     * 1.记录当前任务ID
     * 2.清理执行完成(不管成功还是失败)的任务ID
     * 3.清理任务ID集合(线程池初始化时)
     * 4.检查任务ID集合中的任务，是否已经是终态，如果是终态，则进行清理。此情况是由于在清理执行完成的任务ID时失败导致。
     */
    @Override
    public int performTakeEffectActivity() {

        int successSubmitTaskCount = 0;
        log.info("优惠活动线程池信息:{}", executor.toString());
        // 线程池队列预警阈值
        int alertQueueSize = activityPoolConfiguration.getAlertQueueSize();
        // 每批次查询数量
        int maxLimit = activityPoolConfiguration.getMaxLimit();
        // 最大执行次数
        int maxExecuteTimes = activityPoolConfiguration.getMaxExecuteTimes();

        /*
            查询活动
            1.当前时间大于等于"nextTryTime"
            2.执行状态为待执行或失败
            3.执行次数小于最大执行次数
         */
        List<ActivityTaskEntity> todoTaskList = activityTaskMapper.getTodoList(DateUtil.getCurrentTime(), maxExecuteTimes, maxLimit);

        log.info("优惠活动生产者拉取任务，任务数size={}", todoTaskList.size());
        if (CollectionUtils.isNotEmpty(todoTaskList)) {
            // 批量修改状态
            List<Long> idList = todoTaskList.stream().map(ActivityTaskEntity::getId).collect(Collectors.toList());
            activityTaskMapper.updateExecResultAndLatestInRunningTime(ActivityExecResultEnum.RUNNING.getKey(), DateUtil.getCurrentTime(), idList);

            for (ActivityTaskEntity taskEntity : todoTaskList) {
                log.info("taskId={}，准备放入线程池", taskEntity.getId());
                try {
                    executor.execute(new ActivityTaskWrapper(this, taskEntity));
                    successSubmitTaskCount++;
                    log.info("taskId={}，已经放入线程池", taskEntity.getId());
                } catch (RejectedExecutionException e) {
                    log.error("taskId=" + taskEntity.getId() + "任务提交失败", e);
                }
            }
        }

        // 超时重置，在系统重启的时候，如果线程池中仍有未完成的任务，会导致状态一直锁死在"执行中"，需重置状态。
        int resetCount = activityTaskMapper.resetPooledTask(ActivityExecResultEnum.RUNNING.getKey(), ActivityExecResultEnum.WAIT.getKey(), ActivityExecResultEnum.FAILURE.getKey(), DateUtil.getCurrentTimeWithMinusMinutes(30));
        log.info("优惠活动生产者重置任务数,resetCount={}", resetCount);

        /*
           提交线程池队列任务数量预警
         */
        int size = executor.getQueue().size();
        log.info("优惠活动线程池队列任务数,size={}", size);
        if (size > alertQueueSize) {
            // 一小时内最多通知一次
            if (redisTemplate.opsForValue().setIfAbsent("c:d:r:activity:pool:size", "0", 1, TimeUnit.HOURS) == Boolean.TRUE) {
                SysAlarmInfo sysAlarmInfo = new SysAlarmInfo();
                sysAlarmInfo.setModuleCode("activity");
                sysAlarmInfo.setServiceCode("优惠活动线程池队列大小监控(activity-queue-size)");
                sysAlarmInfo.setMessage("alertQueueSize=" + alertQueueSize + ",executor.getQueue().size()=" + size);
                sysAlarmService.add(sysAlarmInfo);
            }
        }

        /*
          超过重试次数仍未成功预警
         */
        int failureTaskCount = activityTaskMapper.getFailureTaskCount(maxExecuteTimes);
        log.info("优惠活动任务失败数,failureTaskCount={}", failureTaskCount);
        if (failureTaskCount > 0) {
            // 一天内最多通知一次
            if (redisTemplate.opsForValue().setIfAbsent("c:d:r:activity:fail:taskCount", "0", 1, TimeUnit.DAYS) == Boolean.TRUE) {
                SysAlarmInfo sysAlarmInfo = new SysAlarmInfo();
                sysAlarmInfo.setModuleCode("activity");
                sysAlarmInfo.setServiceCode("优惠活动失败任务数提示(subsidy-deduct-fail-task-count");
                sysAlarmInfo.setMessage("failureTaskCount=" + failureTaskCount);
                sysAlarmService.add(sysAlarmInfo);
            }
        }

        return successSubmitTaskCount;
    }

    public void performTakeEffectActivityDo(ActivityTaskEntity taskEntity) {
        Byte execResult;
        String resultMsg;
        try {
            long currentTime = DateUtil.getCurrentTime();
            Long activityId = taskEntity.getActivityId();
            ActivityEntity sEn = new ActivityEntity();
            sEn.setId(activityId);
            ActivityEntity activityEntity = activityMapper.selectByPrimaryKey(sEn);
            String merchantId = activityEntity.getMerchantId();
            // 待生效-->生效中
            if (taskEntity.getTaskType().equals(ActivityTaskTypeEnum.INTERMEDIATE.getKey())) {
                if (activityEntity.getIsActive().equals(IsActiveEnum.YES.getKey())
                        && currentTime >= activityEntity.getActivityStartTime()
                        && currentTime <= activityEntity.getActivityEndTime()) { // 启用状态为启用，并且当前时间落在活动开始与结束区间
                    List<ActivityStoreEntity> storeEntities = activityStoreMapper.findByActivityId(activityId);
                    List<String> storeIdList = storeEntities.stream().map(ActivityStoreEntity::getStoreId).collect(Collectors.toList());
                    if (CollectionUtils.isNotEmpty(storeIdList)) { // 指定挡口
                        List<Long> ids = snowflakeIdService.nextIds(storeIdList.size());
                        List<ActivityActiveStoreEntity> activeStoreEntities = new ArrayList<>(storeIdList.size());
                        for (int i = 0; i < storeIdList.size(); i++) {
                            ActivityActiveStoreEntity activeStoreEntity = new ActivityActiveStoreEntity();
                            activeStoreEntity.setId(ids.get(i));
                            activeStoreEntity.setMerchantId(merchantId);
                            activeStoreEntity.setActivityId(activityId);
                            activeStoreEntity.setStoreId(storeIdList.get(i));
                            activeStoreEntities.add(activeStoreEntity);
                        }
                        activityActiveStoredMapper.insertList(activeStoreEntities);
                    } else { // 不限
                        activityActiveStoredMapper.insert(activitySupport.getUnLimitActiveStoreEn(merchantId, activityId));
                    }
                }
                activityMapper.modifyActivityStatus(activityId, ActivityStatusEnum.ACTIVE.getKey());
            }
            // 生效中-->已终止
            if (taskEntity.getTaskType().equals(ActivityTaskTypeEnum.TERMINAL.getKey())) {
                ActivityTaskEntity interTaskEntity = activityTaskMapper.getPreTask(activityId, ActivityTaskTypeEnum.INTERMEDIATE.getKey());
                if (Objects.nonNull(interTaskEntity)) {
                    // ERR_100018("前置未执行")
                    throw new PromoException(PromoMessageEnum.ERR_430018);
                }
                if (activityEntity.getStatus().equals(IsActiveEnum.YES.getKey())) { // 启用
                    activityActiveStoredMapper.deleteByActivityId(activityId);
                }
                activityMapper.modifyActivityStatus(activityId, ActivityStatusEnum.FINAL.getKey());
            }

            execResult = ActivityExecResultEnum.SUCCESS.getKey();
            resultMsg = "OK";
        } catch (Throwable e) {
            log.error("优惠活动执行异常", e);
            execResult = ActivityExecResultEnum.FAILURE.getKey();
            String stackTrace = ExceptionUtils.getStackTrace(e);
            if (stackTrace.length() > 60035) { // resultMsg text 最大字符数65535，这里只设置60035
                resultMsg = stackTrace.substring(0, 60035);
            } else {
                resultMsg = stackTrace;
            }
        }

        taskEntity.setExecuteTimes(taskEntity.getExecuteTimes() + 1); // 每次执行，执行次数 +1

        // 如果失败的话，下次重试时间等于当前时间 + N分钟(N = 执行次数 * 2)
        taskEntity.setNextTryTime(DateUtil.getCurrentTimeWithMinusMinutes(taskEntity.getExecuteTimes() * 2));
        // 同步结果到 "活动任务表"
        taskEntity.setExecResult(execResult);
        taskEntity.setResultMsg(resultMsg);
        activityTaskMapper.updateByPrimaryKeySelective(taskEntity);
    }


    /**
     * 查询活动，创建任务
     * 1.当前时间大于等于"活动开始时间"的活动，活动状态为待生效
     * 2.当前时间大于等于"活动结束时间"的活动，活动状态为生效中
     */
    @Transactional
    public void createActivityStartTask() {

        List<ActivityEntity> todoTaskList = activityMapper.getActivityStartTodoTask(DateUtil.getCurrentTime(), 0L);

        while (CollectionUtils.isNotEmpty(todoTaskList)) {
            Long activityId = createTask(todoTaskList, ActivityTaskTypeEnum.INTERMEDIATE.getKey());
            activityMapper.updateTraceStartStatus(todoTaskList.stream().map(ActivityEntity::getId).collect(Collectors.toList()));
            todoTaskList = activityMapper.getActivityStartTodoTask(DateUtil.getCurrentTime(), activityId);
        }
    }

    @Transactional
    public void createActivityEndTask() {

        List<ActivityEntity> todoTaskList = activityMapper.getActivityEndTodoTask(DateUtil.getCurrentTime(), 0L);

        while (CollectionUtils.isNotEmpty(todoTaskList)) {
            Long activityId = createTask(todoTaskList, ActivityTaskTypeEnum.INTERMEDIATE.getKey());
            activityMapper.updateTraceEndStatus(todoTaskList.stream().map(ActivityEntity::getId).collect(Collectors.toList()));
            todoTaskList = activityMapper.getActivityEndTodoTask(DateUtil.getCurrentTime(), activityId);
        }
    }

    private Long createTask(List<ActivityEntity> todoTaskList, Byte taskType) {
        Long activityId = null;
        List<ActivityTaskEntity> taskEntities = new ArrayList<>(todoTaskList.size());
        List<Long> ids = snowflakeIdService.nextIds(todoTaskList.size());
        for (int i = 0; i < todoTaskList.size(); i++) {
            ActivityEntity activityEntity = todoTaskList.get(i);

            ActivityTaskEntity taskEntity = new ActivityTaskEntity();
            taskEntity.setId(ids.get(i));
            taskEntity.setActivityId(activityEntity.getId());
            taskEntity.setTaskType(taskType);
            taskEntity.setNextTryTime(DateUtil.getCurrentTime());
            taskEntities.add(taskEntity);
            activityId = activityEntity.getId();
        }
        activityTaskMapper.insertList(taskEntities);
        return activityId;
    }




    /**
     * // pool-poolNumber-thread-threadNumber
     * 线程池工厂
     * 1.根据业务名称重新定义线程名称
     * 2.自定义未捕获异常处理器，防止异常泄漏
     */
    private final ThreadFactory threadFactory = new ThreadFactoryBuilder()
            .setNameFormat("pool-activity-thread-%d")
            .setUncaughtExceptionHandler((t, e) -> {
                log.info("活动状态变更异常线程:{}", t);
                log.error("活动状态变更未捕获异常", e);
            })
            .build();

    private void createPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, int queueSize) {
        if (maximumPoolSize < corePoolSize) {
            maximumPoolSize = corePoolSize;
        }
        executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.SECONDS, new ArrayBlockingQueue<>(queueSize),
                threadFactory, new ThreadPoolExecutor.AbortPolicy()) {
            @Override
            protected void beforeExecute(Thread t, Runnable r) {
                try {
                    ActivityTaskWrapper wrapper = (ActivityTaskWrapper) r;
                    ActivityTaskEntity taskEntity = wrapper.getTaskEntity();
                    activityTaskMapper.recordLatestStartTime(taskEntity.getId(), DateUtil.getCurrentTime());
                } catch (Throwable e) {
                    log.error("记录任务执行开始时间异常", e);
                }
            }

            @Override
            protected void afterExecute(Runnable r, Throwable t) {
                try {
                    ActivityTaskWrapper wrapper = (ActivityTaskWrapper) r;
                    ActivityTaskEntity taskEntity = wrapper.getTaskEntity();
                    activityTaskMapper.recordLatestEndTime(taskEntity.getId(), DateUtil.getCurrentTime());
                } catch (Throwable e) {
                    log.error("记录任务执行结束时间异常", e);
                }
            }
        };
    }

    private void createPool(ActivityPoolConfiguration activityPoolConfiguration) {
        int corePoolSize = activityPoolConfiguration.getCorePoolSize();
        int maximumPoolSize = activityPoolConfiguration.getMaximumPoolSize();
        long keepAliveTime = activityPoolConfiguration.getKeepAliveTime();
        int queueSize = activityPoolConfiguration.getQueueSize();

        queueSizePreviousVersion = queueSize;

        createPool(corePoolSize, maximumPoolSize, keepAliveTime, queueSize);
    }

    /**
     * 用于记录线程池内部队列大小的上一个版本。
     */
    private int queueSizePreviousVersion;

    @EventListener
    public void listen(RefreshScopeRefreshedEvent event) throws InterruptedException {
        log.info("event={}", event.toString());
        int corePoolSize = activityPoolConfiguration.getCorePoolSize();
        int maximumPoolSize = activityPoolConfiguration.getMaximumPoolSize();
        long keepAliveTime = activityPoolConfiguration.getKeepAliveTime();
        int queueSize = activityPoolConfiguration.getQueueSize();

        /*
           线程池内部队列不提供外部设置队列大小的方法，只能先停止线程池，再重新创建
         */
        if (queueSize != queueSizePreviousVersion) {
            queueSizePreviousVersion = queueSize;
            executor.shutdown();
            while (true) {
                boolean flag = executor.awaitTermination(1, TimeUnit.MINUTES);
                if (flag) {
                    break;
                }
            }
            createPool(corePoolSize, maximumPoolSize, keepAliveTime, queueSize);
        } else {
            executor.setCorePoolSize(corePoolSize);
            executor.setMaximumPoolSize(maximumPoolSize);
            executor.setKeepAliveTime(keepAliveTime, TimeUnit.MILLISECONDS);
        }
        log.info("corePoolSize={}, maximumPoolSize={}, keepAliveTime={}, queueSize={}", corePoolSize, maximumPoolSize, keepAliveTime, queueSize);
    }

}
